package com.fafu.mq.mqserver.core;

import ch.qos.logback.core.pattern.color.BlueCompositeConverter;
import com.fafu.mq.common.Consumer;
import com.fafu.mq.common.ConsumerEnv;
import com.fafu.mq.common.MqException;
import com.fafu.mq.mqserver.VirtualHost;

import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

/**
 *
 */

public class ConsumerManager {
    // 持有上层的 VirtualHost 对象的引用, 来操作数据
    private VirtualHost parent;
    // 线程池, 负责执行回调任务
    private ExecutorService worksPool = Executors.newFixedThreadPool(4);
    // "令牌" 阻塞队列
    private BlockingDeque<String> tokenQueue = new LinkedBlockingDeque<>();
    // 扫描线程
    Thread scannerThread = null;

    public ConsumerManager(VirtualHost p) {
        this.parent = p;
        scannerThread = new Thread(() ->{
            while (true) {
                try {
                    // 获取令牌
                    String queueName = tokenQueue.take();
                    // 找到队列
                    MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("[ConsumerManager] 取到令牌后, 发现队列不存在! queueName = " + queueName);
                    }
                    // 从这个消息消费一个队列
                    synchronized (queue){
                        consumeMessage(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        // 把线程设置成后台线程
        scannerThread.setDaemon(true);
        scannerThread.start();

    }

    // 这个方法的调用时机就是发现消息的时候
    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        // 找到对应的队列
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if (queue == null) {
            throw new MqException("[ConsumerManager] 队列不存在! queueName = " + queueName);
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag,queueName,autoAck,consumer);
        synchronized (queue) {
            queue.addConsumerEnv(consumerEnv);
            // 如果队列已经有一些消息了, 需要立刻消费掉
            int n = parent.getMemoryDataCenter().getMessageCount(queueName);
            for (int i = 0; i < n; i++) {
                // 每调用一次消费掉一个消息
                consumeMessage(queue);
            }
        }
    }

    private void consumeMessage(MSGQueue queue) {
        // 1. 按照轮询方式, 找到一个消费者出来
        ConsumerEnv luckyDog = queue.chooseConsumer();
        if (luckyDog == null) {
            // 当前队列没有消费者, 暂时不消费
            return;
        }
        // 2. 从队列取出一个消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if (message == null) {
            // 当前队列没有消息, 不需要进行消费
            return;
        }
        // 3. 把消息带到消费者的回调方法中, 调用线程池
        worksPool.submit(() ->{
            try {
                // 1. 把消息放在待确认队列中, 必须在回调前进行
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(),message);
                // 2. 进行回调
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(),message.getBasicProperties(),message.getBody());
                // 3. 判断是自动应答还是手动应答, 自动应答直接删
                // 手动应答, 后续交给 basicAck 来处理
                if (luckyDog.isAutoAck()) {
                    // 1) 删除硬盘的消息
                    if (message.getDeliverMode() == 2) {
                        parent.getDiskDataCenter().deleteMessage(queue,message);
                    }
                    // 2) 删除待确认集合的消息
                    parent.getMemoryDataCenter().deleteMessageWaitAck(queue.getName(), message.getMessageId());
                    // 3) 删除内存中的消息
                    parent.getMemoryDataCenter().deleteMessage(message.getMessageId());
                    System.out.println("[ConsumerManager] 消息被成功消费了! queueName = " + queue.getName());
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }
}
